package com.thiswhyme.stream.rabbit.consumer2.service;

import com.rabbitmq.client.Channel;
import com.thiswhyme.stream.rabbit.consumer2.BizService.InputChannel;
import com.thiswhyme.stream.rabbit.receive.AbstractRetryReceiver;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;

import java.io.IOException;

/**
 * @description:
 * @author: thiswhyme
 * @date: 2021/1/29
 */
@Slf4j
@Component
public class DemoReceiver extends AbstractRetryReceiver {

    @Override
    public void doIfRetryFail(Object bizParams) {
        log.info("DemoReceiver.INPUT_CHANNEL_MANUAL消息，重试3次后仍然失败，业务参数：{}", bizParams);
    }

    @StreamListener(InputChannel.INPUT_CHANNEL_MANUAL)
    @Override
    public void onMessage(Message<String> message) {
        Channel channel = (Channel) message.getHeaders().get(AmqpHeaders.CHANNEL);
        Long deliveryTag = (Long) message.getHeaders().get(AmqpHeaders.DELIVERY_TAG);
        try {
            if("true".equals(message.getPayload())) {
                log.info("DemoReceiver.INPUT_CHANNEL_MANUAL消息，消息被消费，内容：{}", message.getPayload());
            } else if("true1".equals(message.getPayload())) {
                retry(message.getHeaders(), message.getPayload(), 3, "重试3次后仍然消费失败", null);
            }
        } catch (Exception e) {
            doIfRetryFail("消费过程中存在异常" + e.getMessage());
        } finally {
            try {
                channel.basicAck(deliveryTag, false);
            } catch (IOException e) {
                log.warn("DemoReceiver.INPUT_CHANNEL_MANUAL消息，消息ACK失败，内容：{}", message.getPayload());
            }
        }
    }
}
